// Copyright (C) 2012 Yury Bychenkov (http://bychenkov.com).
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
//  - Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//  - Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
//  - Neither the name of Yury Bychenkov nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

#include "logchunk.h"
#include <fstream>

Logger LogChunk::logger("LogChunk");

LogChunk* LogChunk::open(const char* file, bool readOnly) {
	Header header;

	ifstream is(file, ios::binary | ios::in);
	if (!is.is_open()) {
		logger.error("File ", file, " doesn't exist");
		return 0;
	}
	if (sizeof(Header) != is.read((char*) &header, sizeof(Header)).gcount()) {
		logger.error("Can't load the header: ", file);
		return 0;
	}
	is.close();

	if (header.magic != MAGIC) {
		logger.error("MAGIC mismatch: ", file);
		return 0;
	}

	uint64 size = header.headerSize + header.blockSize * (uint64) header.blockCapacity;

	MappedFile* mappedFile = MappedFile::open(file, size, readOnly);
	if (mappedFile == 0) {
		logger.error("Can't map file ", file , "to memory");
		return 0;
	}

	LogChunk* chunk = new LogChunk(mappedFile);
	if (chunk->mappedFile == 0) {
		delete [] chunk;
		return 0;
	}

	return chunk;
}



LogChunk* LogChunk::create(const char* file, uint columnCount, uint blockSize, uint blockCapacity) {
	uint32 headerSize = (sizeof(Header) + columnCount * sizeof(uint32) + 4095) & ~4095;

	MappedFile* mappedFile = MappedFile::open(file, headerSize + blockSize * (uint64) blockCapacity, false);
	if (mappedFile == 0) {
		logger.error("Can't map file ", file , "to memory");
		return 0;
	}
	

	Header* header = (Header*) mappedFile->data();

	memset(header, 0, headerSize);

	header->magic = MAGIC;
	header->headerSize = headerSize;
	header->blockSize = blockSize;
	header->blockCapacity = blockCapacity;
	header->bucketCount = columnCount + 1;

	for (uint i = 0; i < columnCount + 1; i++) {
		header->bucketTails[i] = blockSize * i + BLOCK_HEADER_SIZE;
	}

	LogChunk* chunk = new LogChunk(mappedFile);
	if (chunk->mappedFile == 0) {
		delete [] chunk;
		return 0;
	}

	return chunk;
}

LogChunk::LogChunk(MappedFile* mappedFile_):
	mappedFile(mappedFile_),
	buckets(0)
{
	header = (Header*) mappedFile->data();

	headerSize = header->headerSize;
	blockSize = header->blockSize;
	blockCapacity = header->blockCapacity;
	bucketCount = header->bucketCount;

	buckets = new Bucket[bucketCount];

	data = mappedFile->data() + headerSize;

   	blockCount = 0; 
	for (uint i = 0; i < bucketCount; i++) {
		uint32 offset = header->bucketTails[i];
		blockCount = max(blockCount, offset / blockSize + 1);

		Bucket& bucket = buckets[i];
		bucket.data = data + (offset / blockSize) * blockSize;
		bucket.offset = offset % blockSize;
		bucket.size = 0;
	}
}

bool LogChunk::isOpened() const {
	return data != 0;
}

void LogChunk::close() {
	if (mappedFile == 0) {
		return;
	}

	header = 0;
	data = 0;
	delete mappedFile;
	mappedFile = 0;
	delete [] buckets;
	buckets = 0;
}

LogChunk::~LogChunk() {
	close();
}

void LogChunk::begin() {
	bucketIndex = 0;
	hashCode = 0;
}

bool LogChunk::add(const void* value, uint valueSize) {
	if (bucketIndex >= bucketCount) {
		return false;
	} 

	Bucket& bucket = buckets[bucketIndex++];
	if (bucket.offset + valueSize > blockSize) {
		if (valueSize + BLOCK_HEADER_SIZE >= blockSize) {
			return false;
		}

		if (blockCount >= blockCapacity) {
			return false;
		}

		uint8* next = data + blockCount * blockSize;
		memset(next, 0, blockSize);

		((uint32*) bucket.data)[0] = (uint32) bucket.offset;
		((uint32*) bucket.data)[1] = (uint32) blockCount;

		bucket.data = next;
		bucket.offset = BLOCK_HEADER_SIZE;

		lock.lock();
		header->bucketTails[bucketIndex - 1] = (uint32) (blockCount * blockSize + BLOCK_HEADER_SIZE);
		lock.unlock();

		blockCount++;
	}

	uint8* dst = bucket.data + bucket.offset;
	for (uint i = 0; i < valueSize; i++) {
		const uint8 b = ((uint8*) value)[i];
		dst[i] = b;
		hashCode = 31 * hashCode + b;
	}

	bucket.size = valueSize;

	return true;
}

bool LogChunk::commit() {
	if (bucketIndex != bucketCount - 1) {
		return false;
	}

	uint32 hc = hashCode == 0? 1: hashCode;

	if (!add(&hc, sizeof(hc))) {
		return false;
	}

	lock.lock();
	for (uint i = 0; i < bucketCount; i++) {
		buckets[i].offset += buckets[i].size;
		header->bucketTails[i] = (uint32) (buckets[i].data - data) + buckets[i].offset;
	}
	lock.unlock();

	return true;
}

uint64 LogChunk::estimateColumnSize(uint columnIndex) const {
	if (columnIndex >= bucketCount) {
		return 0;
	}

	lock.lock();
	uint tail = header->bucketTails[columnIndex];
	lock.unlock();

	uint blockIndex = columnIndex;
	uint lastBlockIndex = (tail - BLOCK_HEADER_SIZE) / blockSize;
	uint64 size = 0;

	while (blockIndex != lastBlockIndex) {
		uint32* p = (uint32*) (data + blockIndex * blockSize);
		size += p[0] - BLOCK_HEADER_SIZE;

		blockIndex = p[1];
	}

	size += tail - blockIndex * blockSize - BLOCK_HEADER_SIZE;


	return size;
}


LogChunk::Iterator::Iterator(const LogChunk& chunk_, uint bucketIndex):
	chunk(&chunk_),
	value(0),
	valueIndex(0)
{
	chunk->lock.lock();
	tail = chunk->header->bucketTails[bucketIndex];
	chunk->lock.unlock();

	data = chunk->data + bucketIndex * chunk->blockSize;
	size = ((uint32*) data)[0];
	size = min(size == 0 ? chunk->blockSize : size, tail - bucketIndex * chunk->blockSize);
		
	offset = BLOCK_HEADER_SIZE;
}

bool LogChunk::Iterator::skipBlock() {
	uint32 nextBlock = ((uint32*) data)[1];
	if (nextBlock == 0) {
		return false;
	}

//cout << nextBlock << "|" << (tail - nextBlock * chunk->blockSize) << " ";
	data = chunk->data + nextBlock * chunk->blockSize;

	size = ((uint32*) data)[0];
	size = min(size == 0 ? chunk->blockSize : size, tail - nextBlock * chunk->blockSize);

	offset = BLOCK_HEADER_SIZE;

	return offset < size;
}
